package com.behavioranalysis.flinkprogram.flink.ad;

import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.source.RichSourceFunction;

import java.sql.Connection;
import java.sql.DriverManager;
import java.sql.PreparedStatement;
import java.sql.ResultSet;

public class BlacklistSource extends RichSourceFunction<String> {

    private Connection conn;
    private PreparedStatement pstm;

    @Override
    public void open(Configuration parameters) throws Exception {
        super.open(parameters);
        String driver = "com.mysql.jdbc.Driver";
        String url = "jdbc:mysql://localhost:3306/spark_project";
        String username = "root";
        String password = "Yangsha130333";
        //1.加载驱动
        Class.forName(driver);
        //2.创建连接
        conn = DriverManager.getConnection(url, username, password);
        //3.获得执行语句
        String sql = "select * from ad_blacklist";
        pstm = conn.prepareStatement(sql);
    }

    @Override
    public void run(SourceContext<String> sourceContext) throws Exception {
        try {
            //4.执行查询，封装数据
            ResultSet resultSet = pstm.executeQuery();
            while (resultSet.next()) {
                String user_id =
                        String.valueOf(resultSet.getInt("user_id"));
                sourceContext.collect(user_id);
            }
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    @Override
    public void cancel() {

    }

    @Override
    public void close() throws Exception {
        super.close();
        //5.关闭连接和释放资源
        super.close();
        if (conn != null) {
            conn.close();
        }
        if (pstm != null) {
            pstm.close();
        }
    }
}
